package com.zx.learn.flink.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.io.*;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;

/**
 * @author Ken.Zheng
 * @date 2023/06/03
 **/
@Slf4j
public class NcMockServer {

    // 定义端口号
    public static final int PORT = 9000;

    private static final int defaultGapMiniSec = 50;

    /**
     * 通过NC产生数据
     *
     * @param dataList 数据列表，多条间间隔默认50毫秒.
     * @throws IOException 输出数据异常.
     */
    public static void generateData(List<String> dataList) throws IOException {
        generateData(dataList, defaultGapMiniSec);
    }

    /**
     * 通过NC产生数据
     *
     * @param dataList          数据列表
     * @param defaultGapMiniSec 生产数据的间隔
     * @throws IOException 输出数据异常.
     */
    public static void generateData(List<String> dataList, int defaultGapMiniSec) throws IOException {
        if (CollectionUtils.isEmpty(dataList)) {
            log.warn("nc mock数据为空");
            return;
        }

        if (defaultGapMiniSec < 50) {
            defaultGapMiniSec = defaultGapMiniSec;
        }

        log.info("nc mock数据生成数据线程启动");
        new MockDataThread(dataList, defaultGapMiniSec).start();
        log.info("nc mock数据生成数据线程启动完成");

    }


    // 监控连接程序是否断开 线程类
    static class MockDataThread extends Thread {

        private List<String> dataList;

        private int sleepGapMiniSecond;

        public MockDataThread(List<String> dataList, int sleepGapMiniSecond) {
            this.dataList = dataList;
            this.sleepGapMiniSecond = sleepGapMiniSecond;
        }

        @Override
        public void run() {
            try {
                log.info("进入数据生成线程");
                ServerSocket server = new ServerSocket(PORT);
                //2秒
                // server.setSoTimeout(2000);
                Socket socket = server.accept();
                //获得远程服务器的地址
                InetAddress netAddress = server.getInetAddress();
                //获得远程服务器的IP地址
                String netIp = netAddress.getHostAddress();
                log.info("ip{}:", netIp);

                log.info("[{}]{}已建立连接！开始异步导入数据，大小：{}，间隔毫秒数:{}",
                        LocalDateTime.now(), socket.getInetAddress(), dataList.size(), sleepGapMiniSecond);

                //输出流
                OutputStream outputStream = socket.getOutputStream();

                dataList.stream().forEach(str -> {
                            try {
                                outputStream.write(str.getBytes());
                                log.debug("nc input{}", str);
                                outputStream.flush();
                                Thread.sleep(sleepGapMiniSecond);
                            } catch (IOException | InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        }
                );

                Thread.sleep(1*1000L);
                //断开
                socket.sendUrgentData(0xFF);
                socket.shutdownOutput();
                socket.close();
            } catch (IOException e) {
                log.error("连接已关闭！", e);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        log.info("生成NC数据");
        List<String> ncData = DataUtils.getData("wordcount.txt");
        log.debug("data:{}", Arrays.asList(ncData));
        NcMockServer.generateData(ncData);

        Socket clent = new Socket("localhost",PORT);
        BufferedReader br = new BufferedReader(new InputStreamReader(clent.getInputStream()));
        String s = "";
        while ((s = br.readLine()) != null) {
            System.out.println("客户端读:"+s);
        }
        clent.close();
    }

}
